package com.im.consumer.coder;

import com.google.protobuf.InvalidProtocolBufferException;
import com.im.consumer.message.Message;
import com.im.consumer.message.proto.PackData;
import com.im.consumer.message.proto.PackHeader;
import org.apache.kafka.common.serialization.Deserializer;


import java.nio.ByteBuffer;

public class MessageDeserializer implements Deserializer<Message> {

    @Override
    public Message deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }

        try {
            // 使用 ByteBuffer 来解析消息
            ByteBuffer buffer = ByteBuffer.wrap(data);

            // 读取并解析包边界、版本、包头长度、包体长度
            int packetBoundary = buffer.getInt();  // 获取包边界（4字节）
            int version = buffer.getInt();         // 获取版本（4字节）
            int headerLength = buffer.getInt();    // 获取包头长度（4字节）
            int bodyLength = buffer.getInt();      // 获取包体长度（4字节）

            // 检查是否有足够的字节数据
            if (buffer.remaining() < headerLength + bodyLength) {
                throw new IllegalArgumentException("没有这么长的长度");
            }

            // 根据包头长度读取包头数据
            byte[] headerBytes = new byte[headerLength];
            buffer.get(headerBytes); // 读取包头字节数据
            PackHeader.Header header = PackHeader.Header.parseFrom(headerBytes);  // 使用 Protobuf 解析包头

            // 根据包体长度读取包体数据
            byte[] bodyBytes = new byte[bodyLength];
            buffer.get(bodyBytes);  // 读取包体字节数据
            PackData.Body body = PackData.Body.parseFrom(bodyBytes);  // 使用 Protobuf 解析包体

            // 构造 Message 对象并设置各个字段
            Message message = new Message();
            message.setPacketBoundary(packetBoundary);
            message.setVersion(version);
            message.setPacketHeaderLength(headerLength);
            message.setPacketDataLength(bodyLength);
            message.setHeader(header);
            message.setBody(body);

            return message;
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException("Error while deserializing Message", e);
        }
    }
}
